﻿using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;



namespace ConsoleAppKafkaTest
{
    class KafkaProducerDemo
    {
        public static void TestProducer()
        {
            ProducerConfig config = new ProducerConfig();
            config.BootstrapServers = "127.0.0.1:9092";

            ProducerBuilder<string, object> builder = new ProducerBuilder<string, object>(config);
            builder.SetValueSerializer(new KafKaConverter());
            builder.SetErrorHandler((p, error) =>
            {
                Console.WriteLine($"{p.Name}  IsBrokerError：{error.IsBrokerError}  IsFatal：{error.IsFatal} IsLocalError:{error.IsLocalError}");
            });
            IProducer<string, object> producer = builder.Build();


            Console.WriteLine("按确认键发送kafka信息：");

            //TimeSpan flushTime = TimeSpan.FromMilliseconds(200);
            while (true)
            {
                ConsoleKeyInfo keyinfo = Console.ReadKey();
                if (keyinfo.Key != ConsoleKey.Enter)
                {
                    break;
                }
                try
                {
                    producer.Produce("hah", new Message<string, object>()
                    {
                        Key = "测试",
                        Value = "123value" + DateTime.Now
                    });
                    //, r => { Console.WriteLine($"发送数据结果：{!r.Error.IsError}"); }
                    //int flushNu = producer.Flush(flushTime);

                    Console.WriteLine($"{DateTime.Now}:已经发送.发送数量：");
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"发送发生异常：{ex.Message}");
                }
            }

        }
    }

    class KafKaConverter : ISerializer<object>
    {
        public byte[] Serialize(object data, SerializationContext context)
        {
            return Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(data));
        }
    }
}
